package org.onion.pattern.event.springboot.rocketmq.consume;

import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.acl.common.AclClientRPCHook;
import org.apache.rocketmq.acl.common.SessionCredentials;
import org.apache.rocketmq.client.consumer.listener.MessageListener;
import org.apache.rocketmq.client.exception.MQClientException;
import org.onion.pattern.event.consumer.EventConsumeHandler;
import org.onion.pattern.event.spring.io.EventCodecs;
import org.onion.pattern.event.springboot.rocketmq.annotation.RocketMQConsumerGroup;
import org.onion.pattern.event.springboot.rocketmq.annotation.RocketMQEvent;
import org.onion.pattern.event.springboot.rocketmq.annotation.RocketMQEventListener;
import org.onion.pattern.event.springboot.rocketmq.annotation.enums.ListenerMode;
import org.onion.pattern.event.springboot.rocketmq.config.RocketMQEventProperties;
import org.onion.pattern.event.springboot.rocketmq.consume.listener.RocketMessageConcurrentlyListener;
import org.onion.pattern.event.springboot.rocketmq.consume.listener.RocketMessageOrderlyListener;
import org.onion.pattern.event.springboot.rocketmq.topic.ITopicGenerator;
import org.onion.pattern.event.springboot.rocketmq.topic.RocketMQTopic;
import org.springframework.context.ApplicationContext;
import org.springframework.core.annotation.AnnotationUtils;

import java.lang.reflect.Method;
import java.text.MessageFormat;
import java.util.*;
import java.util.stream.Collectors;
import java.util.stream.Stream;

@Slf4j
public class RocketMQMessageConsumer {
    //        private final EventDrivenRocketMQProperties.ConsumerProperties consumerProperties;
    @Getter
    private final Class<?>                 consumerGroupClass;
    @Getter
    private       RocketMQConsumerGroup    consumerGroupAnnotation;
    private       EventConsumeHandler      rocketEventHandler;
    private       RocketMQEventProperties  rocketMQEventConfig;
    @Setter
    @Getter
    private       EventCodecs              eventCodecs;
    @Setter
    @Getter
    private       ITopicGenerator          topicGenerator;
    private       _Consumer                consumer;
    private final TopicAndEventTypeMapping topicAndEventTypeMapping = new TopicAndEventTypeMapping();


    public RocketMQMessageConsumer(EventConsumeHandler rocketEventHandler
//            , DynamicConsumer dynamicConsumer
//            , ConsumerPropertiesFactory consumerPropertiesFactory
            , RocketMQEventProperties rocketMQEventConfig, ApplicationContext applicationContext) {

        this.rocketEventHandler = rocketEventHandler;
        this.consumerGroupClass = rocketEventHandler.consumerGroupClass();
        this.consumerGroupAnnotation = consumerGroupClass.getAnnotation(RocketMQConsumerGroup.class);
        this.rocketMQEventConfig = Objects.requireNonNull(rocketMQEventConfig,"rocketmq 配置不为空");
        this.setTopicGenerator(applicationContext.getBean(ITopicGenerator.class));
//        String dynamicKey = dynamicConsumer != null ? dynamicConsumer.value() : null;
//        this.consumerProperties = consumerPropertiesFactory.create(dynamicKey, getEventListenerAnnotation(), getEventListenerType());
    }


    public void start() {
        try {
            _Consumer consumer = initConsumer();
            consumer.setEnabledAutoCreateTopic(rocketMQEventConfig.getConsumer().isEnabledAutoCreateTopic());
            consumer.setTopicQueueNums(rocketMQEventConfig.getConsumer().getTopicQueueNums());
            consumer.start();
            log.info(MessageFormat.format("【ROCKET事件消费】订阅者启动成功 [{0}:{1}]", consumer.getConsumerGroup(), consumer.getInstanceName()));
        } catch (MQClientException e) {
            log.error(e.getErrorMessage(), e);
        }
    }


    public void close() {
        consumer.shutdown();
    }

    public String getAccessKey() {
        return this.rocketMQEventConfig.getAccessKey();
    }

    public String getSecretKey() {
        return this.rocketMQEventConfig.getSecretKey();
    }


    protected int getConsumeThreadMax() {
        return consumerGroupAnnotation.consumeThreadMax();
    }


    protected String getNamespace() {
        //todo
        return this.rocketMQEventConfig.getDefaultNamespace();
    }


    protected String getNameServer() {
        //todo
        return this.rocketMQEventConfig.getDefaultNameServer();
    }


    protected _Consumer initConsumer() {
        String groupName = RocketMQConsumerGroupUtils.getGroupName(consumerGroupClass);

        if (StringUtils.isNotEmpty(getAccessKey()) && StringUtils.isNotEmpty(getSecretKey())) {
            final AclClientRPCHook aclClientRPCHook = new AclClientRPCHook(new SessionCredentials(getAccessKey(), getSecretKey()));
            consumer = new _Consumer(groupName, aclClientRPCHook);
        } else {
            consumer = new _Consumer(groupName);
        }

        String nameServer = getNameServer();
        String namespace = getNamespace();
        if (consumer.getTopicQueueNums() <= 0) {
            consumer.setTopicQueueNums(16);
        }
        consumer.setNamespace(namespace);
        consumer.setNamesrvAddr(nameServer);
        consumer.setConsumeThreadMax(getConsumeThreadMax());
        if (consumer.getConsumeThreadMax() < consumer.getConsumeThreadMin()) {
            consumer.setConsumeThreadMin(consumer.getConsumeThreadMax());
        }

        consumer.setInstanceName(UUID.randomUUID().toString());
        consumer.setConsumeMessageBatchMaxSize(1);
        consumer.setConsumeFromWhere(consumerGroupAnnotation.consumeFromWhere());
        consumer.setMessageModel(consumerGroupAnnotation.messageModel());
        consumer.setConsumeTimeout(consumerGroupAnnotation.consumeTimeout());

        //腾讯云无法应用该方法
//    MQClientInstance mqClientInstance = MQClientManager
//        .getInstance().getOrCreateMQClientInstance(consumer, null);
//    consumer.setOffsetStore(
//        new CustomRemoteBrokerOffsetStore(mqClientInstance, consumer.getConsumerGroup()));
//    consumer.setConsumeTimestamp(
//        UtilAll.timeMillisToHumanString3(
//            System.currentTimeMillis() + (eventListenerAnnotation.consumeTimestamp())));


        //后面有需要将订阅方法数据抽取解耦单独出去，可以根据自己需要更改订阅方法，如加注解等方式. 找出所有配置有 RocketMQEventListener的方法
        Method[] subscribeMethods =
                Arrays.stream(consumerGroupClass.getDeclaredMethods())
                        .filter(m -> AnnotationUtils.findAnnotation(m, RocketMQEventListener.class) != null)
//                        .filter(x -> x.getName().equals(Properties4RocketMQ.SUBSCRIBE_METHOD))

//            .sorted(
//                (o1, o2) -> {
//                  final Class<?>[] parameterTypes1 = o1.getParameterTypes();
//                  final Class<?>[] parameterTypes2 = o2.getParameterTypes();
//                  Integer i1 = parameterTypes1[0].o1.getName().hashCode();
//                  Integer i2 = o2.getName().hashCode();
//                  return i1.compareTo(i2); // 升序
//                }
//            )
                        .toArray(Method[]::new);

        Map<String, List<String>> topics = new HashMap<>();
        for (Method subscribeMethod : subscribeMethods) {

            final Class<?>[] parameterTypes = subscribeMethod.getParameterTypes(); //获取函数的参数类型

            if (parameterTypes.length > 0) {
                Class<?> parameterClazz = null;
                List<RocketMQEvent> rocketMQEvents = new ArrayList<>();

                for (Class<?> parameterType : parameterTypes) {
                    RocketMQEvent annotation = AnnotationUtils.findAnnotation(parameterType, RocketMQEvent.class);
                    if (annotation != null) {
                        parameterClazz = parameterType;
                        rocketMQEvents.add(annotation);
                    }
                }

                if (rocketMQEvents.isEmpty()) {
                    continue; //方法中都没有 Event参数
                } else if (rocketMQEvents.size() > 1) {
                    throw new IllegalStateException("【ROCKET事件消费】 参数大于 " + rocketMQEvents.size() + "个 方法" + subscribeMethod);
                }

                RocketMQTopic topicModel = this.topicGenerator.generate(this.consumerGroupClass, parameterClazz);
                topics.putIfAbsent(topicModel.getTopic(), new ArrayList<>());
                topics.get(topicModel.getTopic()).add(topicModel.getTag());
                topicAndEventTypeMapping.create(topicModel.getTopic(), topicModel.getTag(), parameterClazz);
//        LegcySubscribeEventTypeFactory
//            .add(parameterType.getName(),
//                parameterType);
            }
        }

        //消费订阅
        for (Map.Entry<String, List<String>> entry : topics.entrySet()) {
            String topic = entry.getKey();
            List<String> filterTags = entry.getValue().stream().filter(StringUtils::isNotBlank).collect(Collectors.toList());

            try {
                if (filterTags.isEmpty()) {
                    consumer.subscribe(topic, "*");
                    continue;
                }

                final Optional<String> isAllSubscribe = filterTags.stream().filter("*"::equals).findAny();
                if (isAllSubscribe.isPresent()) {
                    consumer.subscribe(topic, "*");
                    continue;
                }

                final Stream<String> sortedTopics = filterTags.stream().sorted(
                        (o1, o2) -> {
                            int x = o1.hashCode();
                            int y = o2.hashCode();
                            return Integer.compare(x, y);
                        }
                );

                consumer.subscribe(topic, String.join(
                        " || ",
                        CollectionUtils.emptyIfNull(sortedTopics.collect(Collectors.toList()))));

            } catch (MQClientException e) {
                log.error("【ROCKET事件消费】消费者订阅信息异常", e);
            }
        }

        consumer.setMessageListener(getMessageListener());
        return consumer;
    }


    protected MessageListener getMessageListener() {
        if (consumerGroupAnnotation.listenerMode() == ListenerMode.Concurrently) {
            return new RocketMessageConcurrentlyListener(consumer, rocketEventHandler, eventCodecs, topicAndEventTypeMapping);
        }
        return new RocketMessageOrderlyListener(consumer, rocketEventHandler, eventCodecs, topicAndEventTypeMapping);
    }


}
